package gu.sql2java.manager;

import java.lang.reflect.Array;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.MoreObjects;
import com.google.common.base.Predicates;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableBiMap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.primitives.Ints;

import gu.sql2java.BaseBean;
import gu.sql2java.BaseRow;
import gu.sql2java.ColumnCodec;
import gu.sql2java.Constant;
import gu.sql2java.ForeignKeyMetaData;
import gu.sql2java.IDataSourceConfig;
import gu.sql2java.IndexMetaData;
import gu.sql2java.ListenerContainer;
import gu.sql2java.RowMetaData;
import gu.sql2java.SimpleLog;
import gu.sql2java.TableListener;
import gu.sql2java.TableManager;
import gu.sql2java.ForeignKeyMetaData.ForeignKeyRule;
import gu.sql2java.exception.DaoException;
import gu.sql2java.exception.DataAccessException;
import gu.sql2java.exception.DataRetrievalException;
import gu.sql2java.exception.ObjectRetrievalException;
import gu.sql2java.exception.OptimisticLockingException;
import gu.sql2java.exception.QueueTimeoutException;
import gu.sql2java.exception.RuntimeDaoException;
import gu.sql2java.geometry.GeometryDataCodec;
import gu.sql2java.manager.Manager.AutoKeyRetrieveType;
import gu.sql2java.manager.parser.ParserSupport;

import static com.google.common.base.Preconditions.*;
import static gu.sql2java.SimpleLog.*;
import static gu.sql2java.exception.DaoException.stripSQLException;
import static gu.sql2java.manager.Managers.*;
import static com.google.common.base.MoreObjects.firstNonNull;
import static com.google.common.base.Preconditions.checkNotNull;
import static gu.sql2java.Sql2javaSupport.hasNull;
import static gu.sql2java.Sql2javaSupport.hasNullPk;


/**
 * implementation of {@link TableManager} 
 * @author guyadong
 *
 * @param <B>  java bean type
 */
public class BaseTableManager<B extends BaseBean> implements TableManager<B>,Constant{
    protected final RowMetaData metaData;
    private volatile Manager manager;
    /** lazy load */
	private volatile Map<String,TableListener<BaseBean>> foreignKeyDeleteListeners;
	/** lazy load */
	private volatile ListenerContainerLocal<B> listenerContainer;
	/** lazy load */
	private volatile String generatedkeyStatement;
	private static boolean debug = false;
	protected BaseTableManager(String tablename){
    	metaData = RowMetaData.getMetaData(tablename);
    }

    @Override
    public ListenerContainer<B> getListenerContainer() {
		// double checking
		if(listenerContainer == null){
			synchronized (this) {
				if(listenerContainer == null){
					listenerContainer = new ListenerContainerLocal<B>(getManager().getFireType());
				}
			}
		}
		return listenerContainer;
	}
    @Override
	public IDataSourceConfig getDataSourceConfig(){
    	return getManager().config;
    }
    /**
     * @return map with foreignKey name TO TableListener
     */
    public Map<String, TableListener<BaseBean>> getForeignKeyDeleteListeners(){
    	// double checking
    	if(foreignKeyDeleteListeners == null){
    		synchronized (this) {
    			if(foreignKeyDeleteListeners == null){
    				LinkedHashMap<String, TableListener<BaseBean>> map = Maps.newLinkedHashMap();
    				for(ForeignKeyMetaData fk : metaData.getForeignKeysForListener()){
    					map.put(fk.name, new DeleteRuleListener<BaseBean>(fk.name));
    				}
    				foreignKeyDeleteListeners = Collections.unmodifiableMap(map);
    			}
			}
    	}
    	return foreignKeyDeleteListeners;
    }
	private String getGeneratedkeyStatement(){
		// double check
		if(generatedkeyStatement == null){
			synchronized (this) {
				if(generatedkeyStatement == null){
					generatedkeyStatement = checkNotNull(getManager().getGeneratedkeyStatement(),"INVALID generatedkeyStatement")
						.replaceAll("<TABLE>", metaData.tablename )
						.replaceAll("<KEY>", metaData.columnNameOf(metaData.autoincrementColumnId));
					if(debug){
						log("generatedkeyStatement={}",generatedkeyStatement);
					}
				}
			}
		}
		return generatedkeyStatement;
	}

	@Override
	@SuppressWarnings("unchecked")
    public final B createBean()
    {
    	try {
			return (B) metaData.beanType.newInstance();
		} catch (Exception e) {
			Throwables.throwIfUnchecked(e);
			throw new RuntimeException(e);
		}
    }
	/**
	 * Creates a new B instance.
	 * @param primaryValues values of primary keys
	 * @return B instance OR null if exist null value in primaryValues
	 */
	protected final B createBean(Object... primaryValues)
    {
		checkArgument(primaryValues != null && primaryValues.length== metaData.primaryKeyNames.length,"INVALID primaryValues");
		B bean = createBean();
		int[] pkIds = metaData.primaryKeyIds;
		for(int i=0;i<primaryValues.length;++i){
			int columnId = pkIds[i];
			Object value = primaryValues[i];
			if(null == value) {
				return null;
			}
			try {
				bean.setValue(columnId, value);
			} catch (RuntimeException e) {
				throw new IllegalArgumentException(
						String.format("INVALID primkey type %s for %s.%s,Cased by %s", 
								value.getClass().getName(), 
								metaData.tablename,
								metaData.columnNameOf(columnId),
								e.getMessage()),
						e);
			}
		}
		return bean;
    }
    private void prepareAutoincrement( Connection c,PreparedStatement ps, B bean) throws SQLException
    {
        if (!bean.isModified(metaData.autoincrementColumnId))
        {
            PreparedStatement ps2 = null;
            ResultSet rs = null;
            try {
            	AutoKeyRetrieveType retrieveType = getManager().getGeneratedkeyRetrieveType();
                if(AutoKeyRetrieveType.AUTO.equals(retrieveType)){
                	rs = ps.getGeneratedKeys();
                }else{
                    ps2 = getManager().getStatementCache().prepareStatement(c, getGeneratedkeyStatement(), false, false, null);
                	rs = ps2.executeQuery();
                }
                if(rs.next()) {
                    setColumnValue(bean, 
                    		metaData.autoincrementColumnId, 
                    		getManager().getObject(rs,1,metaData.fieldTypeOf(metaData.autoincrementColumnId)));
                } else {
                    throw new IllegalStateException(logString("ATTENTION: Could not retrieve generated key!(retrieveType:{})",retrieveType));
                }
            } finally {
               getManager().close(ps2, rs);
            }
        }
    }
	//13
    /**
     * Insert the B bean into the database.
     * 
     * @param bean the B bean to be saved
     * @return the inserted bean
     * @throws RuntimeDaoException
     */
    protected B insert(final B bean)
    {
        // mini checks
        if (null == bean || !bean.beModified()) {
            return bean; 
        }
        String productName = this.getManager().getProductName();
        if (!bean.isNew() && !productName.equals(PRODUCT_NAME_PHOENIX)){
            return this.update(bean);
        }

        Connection c = null;
        PreparedStatement ps = null;

		try
        {
            c = this.getConnection();
            final AutoKeyRetrieveType retrieveType = getManager().getGeneratedkeyRetrieveType();
            if(metaData.autoincrementColumnId >= 0){
                /** ignore auto increment  key*/
                bean.resetPrimaryKeysModified();
            }
            //-------------writePreInsert
            if(metaData.autoincrementColumnId >= 0 
            		&& AutoKeyRetrieveType.BEFORE.equals(retrieveType)){
                prepareAutoincrement(c, null, bean);
        	}
            //------------/writePreInsert
            // listener callback
            getListenerContainer().beforeInsert(bean);
            String cmd = productName.equals(PRODUCT_NAME_PHOENIX) ? "UPSERT " : "INSERT ";
            StringBuilder builder = new StringBuilder(cmd + " INTO " + metaData.tablename + " (");
            Iterable<String>modifiedList = Iterables.filter(metaData.columnNames,bean::isModified);
            String fields=Joiner.on(",").join(modifiedList);
            // fields
            builder.append(fields);
            builder.append(") values (");
            // values
            builder.append(fields.replaceAll("[^,]+", "?"));
            builder.append(")");
            if(metaData.autoincrementColumnId >= 0 && AutoKeyRetrieveType.AUTO.equals(retrieveType)){
                ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), false, debug, 
                    cmd, Statement.RETURN_GENERATED_KEYS);
            }else{
                ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), false, debug, 
                    cmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            }
            
            fillPreparedStatement(ps, bean, SEARCH_EXACT, true, true, null, 0);

            ps.executeUpdate();
            //------------------writePostInsert
            if(metaData.autoincrementColumnId >= 0 && !AutoKeyRetrieveType.BEFORE.equals(retrieveType)){
        		prepareAutoincrement(c, ps, bean);
        	}
            //-------------------/writePostInsert
            bean.setNew(false);
            bean.resetIsModified();
            // listener callback
            getListenerContainer().afterInsert(bean);
            return bean;
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(e);
        }
        finally
        {
            // listener callback
            getListenerContainer().done();
            getManager().close(ps);
            freeConnection(c);
        }
    }
    private String makeInsertValuesSql(String cmd,List<String> fields, int lineCount) throws SQLException{
        final AutoKeyRetrieveType retrieveType = getManager().getGeneratedkeyRetrieveType();
        final String valueLine;
        String columnNames = Joiner.on(',').join(fields);
        if(metaData.autoincrementColumnId >= 0 
                && AutoKeyRetrieveType.BEFORE.equals(retrieveType)){
            valueLine = "(" + getGeneratedkeyStatement() + "," + columnNames.replaceAll("[^,]+", "?")+ ")";
            columnNames = metaData.primaryKeyNames[0] + "," + columnNames;
        }else {
            valueLine = "(" + columnNames.replaceAll("[^,]+", "?")+ ")";
        }
        
    	StringBuilder builder = new StringBuilder(cmd + " INTO " + metaData.tablename + " ("+ columnNames +") VALUES ");
        for(int idx=0,endIdx=lineCount;idx<endIdx;++idx){
            if(idx > 0){
                builder.append(",");
            }
            builder.append(valueLine);
        }
        return builder.toString();
    }
    //13-1
    @Override
    public <C extends Iterable<B>>void fastInsert(int[] fieldList,C beans)
    {
        if(null == beans){
            return ;
        }
        Iterable<B> nonullBeans = Iterables.filter(beans,Predicates.notNull());
        if(Iterables.isEmpty(nonullBeans)){
            return ;
        }
        final List<String> fields;
        /* compute column names for insert */
        if(null != fieldList && fieldList.length > 0) {
            for(int columnId:fieldList){
                checkArgument( columnId >=0 && columnId < metaData.columnCount,"INVALID column id %s",columnId);
            }
            Iterable<String> columnNameList = Iterables.transform(Ints.asList(fieldList),metaData.columnFullFieldList::get);
            fields = Lists.newArrayList(columnNameList);
        }else {
            fieldList = metaData.defaultColumnIdList;
            fields = metaData.columnFullFieldList;
        }
        Connection c = null;
        PreparedStatement ps = null;
        
        try
        {
            c = this.getConnection();
            if(metaData.autoincrementColumnId >= 0){
                /** ignore auto increment  key*/
                nonullBeans.forEach(b->b.resetPrimaryKeysModified());
            }
            String productName = this.getManager().getProductName();
            String cmd = productName.equals(PRODUCT_NAME_PHOENIX) ? "UPSERT" : "INSERT";
            if(getManager().isSupportInsertValues()){
                String sql = makeInsertValuesSql(cmd, fields, Iterables.size(nonullBeans));
                ps = getManager().getStatementCache().prepareStatement(c, sql, false, debug, 
                        cmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                int dirtyCount = 0;
                for(BaseBean bean : nonullBeans){
                	dirtyCount = fillPreparedStatement(ps, bean, SEARCH_EXACT, true, false, fieldList, dirtyCount);
                }
                ps.executeUpdate();
            }else {
                String sql = makeInsertValuesSql(cmd, fields, 1);
                ps = getManager().getStatementCache().prepareStatement(c, sql, false, debug, 
                        cmd, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                for(BaseBean bean : nonullBeans){
                    fillPreparedStatement(ps, bean, SEARCH_EXACT, true, false, fieldList, 0);
                    ps.addBatch();
                }
                c.setAutoCommit(false);
                boolean commit = false;
                try {
                    ps.executeBatch();
                    commit = true;
                } finally {
                    if(commit){
                        c.commit();
                    }else {
                        c.rollback();
                    }
                }
            }
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(e);
        }
        finally
        {
            getManager().close(ps);
            freeConnection(c);
        }
    }
    private final Function<String, String> columnExpFun = input->input + "=?";
    
    //14
    /**
     * Update the B bean record in the database according to the changes.
     *
     * @param bean the B bean to be updated
     * @return the updated bean
     * @throws RuntimeDaoException
     */
	@SuppressWarnings("unchecked")
	protected B update(final B bean) throws RuntimeDaoException
    {
        // mini checks
        if (null == bean || !bean.beModified()) {
            return bean;
        }
        if (bean.isNew()){
            return this.insert(bean);
        }
        if(this.getManager().getProductName().equals(PRODUCT_NAME_PHOENIX)){
        	return this.insert(bean);
        }

        Connection c = null;
        PreparedStatement ps = null;

        try
        {
            c = this.getConnection();
            // listener callback
            getListenerContainer().beforeUpdate(bean); 
            Object oldLockValue = null;
            if(metaData.lockColumnType != null){
            	oldLockValue = bean.getValue(metaData.lockColumnName);
        		// lockColumnType is String or Long
        		if(String.class == metaData.lockColumnType){
        			bean.setValue(metaData.lockColumnName,String.valueOf(System.currentTimeMillis())); 
        		}else if(Long.class ==metaData.lockColumnType){
        			bean.setValue(metaData.lockColumnName,System.currentTimeMillis()); 
        		}else{
        			throw new RuntimeException(logString("INVALID LOCK COLUMN TYPE:{},String or Long required",metaData.lockColumnType));
        		}
            }
            StringBuilder builder = new StringBuilder("UPDATE " + metaData.tablename + " SET ");
            builder.append(Joiner.on(",").join(Iterables.transform(Iterables.filter(metaData.columnNames,bean::isModified),columnExpFun)));

            builder.append(" WHERE ");
            builder.append(Joiner.on(" AND ").join(
	            Lists.transform(Arrays.asList(metaData.primaryKeyNames), columnExpFun)));
            if(metaData.lockColumnType != null){
            	if(metaData.primaryKeyNames.length > 0){
            		builder.append(" AND ");
            	}
            	checkArgument(metaData.lockColumnName != null, "NOT DEFINED lock column name");
            	builder.append(metaData.lockColumnName + "=?");
            }
            ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), true, debug, 
                    "update", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            int dirtyCount = this.fillPreparedStatement(ps, bean, SEARCH_EXACT,true, true, null, 0);

            if (dirtyCount == 0) {
            	if(debug){
            		log("The bean to look is not initialized... do not update.");
            	}
                return bean;
            }
            int[] pkIds = metaData.primaryKeyIds;
            for(int i = 0; i<pkIds.length; ++i){
            	setPreparedStatementWithBean(ps, ++dirtyCount, bean, pkIds[i]);
            }
            if(metaData.lockColumnType != null){
            	setPreparedStatement(ps, ++dirtyCount, oldLockValue, metaData.columnIDOf(metaData.lockColumnName));
            }
            int rowCount = ps.executeUpdate();
            if (metaData.lockColumnType != null && rowCount==0) {
                throw new OptimisticLockingException("sql2java.exception.optimisticlock");
            }
            // listener callback
            getListenerContainer().afterUpdate((B)bean.clone()); 
            bean.resetIsModified();

            return bean;
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(e);
        }
        finally
        {
            // listener callback
            getListenerContainer().done();
            getManager().close(ps);
            freeConnection(c);
        }
    }
    
    @Override
    public int countAll()throws RuntimeDaoException{
        return this.countWhere("");
    }

    @Override
    public int countUsingTemplate(B bean)throws RuntimeDaoException{
        return this.countUsingTemplate(bean, SEARCH_EXACT);
    }

    @Override
    public int deleteAll()throws RuntimeDaoException{
        return this.deleteByWhere("");
    }

    @Override
    public B[] loadAll()throws RuntimeDaoException{
        return this.loadByWhere(null);
    }

    @Override
    public int loadAll(TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByWhere(null, action);
    }

    @SuppressWarnings("unchecked")
	@Override
    public B[] loadAll(int startRow, int numRows)throws RuntimeDaoException{
        return this.loadByWhereAsList(null, null, startRow, numRows).toArray((B[]) Array.newInstance(metaData.beanType, 0));
    }

    @Override
    public int loadAll(int startRow, int numRows, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByWhereForAction(null, null, null, startRow, numRows, action);
    }

    @Override
    public List<B> loadAllAsList()throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(null,1, -1, SEARCH_EXACT);
    }

    @Override
    public List<B> loadAllAsList(int startRow, int numRows)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(null, startRow, numRows, SEARCH_EXACT);
    }

    @Override
	public B loadByPrimaryKey(B bean)throws RuntimeDaoException{
		return bean==null ? null : loadByPrimaryKey(bean.primaryValues());
	}

	@Override
	public B loadByPrimaryKeyChecked(B bean)throws RuntimeDaoException,ObjectRetrievalException{
	    return loadByPrimaryKeyChecked(checkNotNull(bean,"bean is null").primaryValues());
	}
	@Override
	public final B loadByPrimaryKeyChecked(Object ...keys)throws RuntimeDaoException,ObjectRetrievalException{
		// {{check parameters
		if(metaData.primaryKeyNames.length == 0){
			throw new UnsupportedOperationException();
		}
	    String[] pkNames = metaData.primaryKeyNames;
	    int[] pkIds = metaData.primaryKeyIds;
		if(hasNull(keys)){
			throw new ObjectRetrievalException(new NullPointerException("primary key must not be null"));
		}
		checkArgument(keys.length == pkNames.length,
				"INVALID ARGUMENT NUM, %s required",pkNames.length);
	    for(int i=0; i<pkNames.length; ++i){
	    	Object key = keys[i];
	    	Class<?> type = metaData.columnTypeOf(pkIds[i]);
	    	checkArgument(type.isAssignableFrom(key.getClass()),
	    			"INVALID type for pk %s,%s required",pkNames[i],type.getName());
	    }
	 // }}check parameters
	    return doLoadByPrimaryKeyChecked(keys);
	}
	
	protected B doLoadByPrimaryKeyChecked(Object ...keys)throws RuntimeDaoException,ObjectRetrievalException{
		if(metaData.primaryKeyNames.length == 0){
			throw new UnsupportedOperationException();
		}
	    String[] pkNames = metaData.primaryKeyNames;
	    int[] pkIds = metaData.primaryKeyIds;
		if(hasNull(keys)){
			throw new ObjectRetrievalException(new NullPointerException("primary key must not be null"));
		}
		checkArgument(keys.length == pkNames.length,
				"INVALID ARGUMENT NUM, %s required",pkNames.length);
	    for(int i=0; i<pkNames.length; ++i){
	    	Object key = keys[i];
	    	Class<?> type = metaData.columnTypeOf(pkIds[i]);
	    	checkArgument(type.isAssignableFrom(key.getClass()),
	    			"INVALID type for pk %s,%s required",pkNames[i],type.getName());
	    }
	    Connection c = null;
	    PreparedStatement ps = null;
	    try
	    {
	        c = this.getConnection();
	        StringBuilder builder = new StringBuilder("SELECT " + metaData.columnFullFields + " FROM " + metaData.tablename + " WHERE ");
	        builder.append(Joiner.on(" AND ").join(
		            Lists.transform(Arrays.asList(pkNames), columnExpFun)));
            ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), false, debug, 
                    "LOAD BY PK", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
			for(int i = 0; i<pkNames.length; ++i){
				setPreparedStatement(ps,  i+1, keys[i], pkIds[i]);
			}
	        ListAction<B> action = new ListAction<>(true);
	        loadByPreparedStatement(ps,null,1,-1,action);
	        List<B> pReturn = action.getList();
	        if (1 == pReturn.size()) {
	            return pReturn.get(0);
	        } else {
	            throw new ObjectRetrievalException();
	        }
	    }
	    catch(ObjectRetrievalException e)
	    {
	        throw e;
	    }
	    catch(SQLException e)
	    {
	        throw new RuntimeDaoException(new DataRetrievalException(e));
	    }
	    finally
	    {
	        this.getManager().close(ps);
	        this.freeConnection(c);
	    }
	}

	@Override
	public B loadByPrimaryKey(Object ...keys)throws RuntimeDaoException{
	    try{
	        return loadByPrimaryKeyChecked(keys);
	    }catch(ObjectRetrievalException e){
	        // not found
	        return null;
	    }
	}

	protected <K> List<B> loadByPks(Collection<K> keys){
		checkState(metaData.primaryKeyCount == 1,"UNSUPPORTED OPERATION");
	    if(null == keys){
	        return Collections.emptyList();
	    }
	    ArrayList<B> list = new ArrayList<B>(keys.size());
	    for(K key:keys){
	        list.add(loadByPrimaryKey(key));
	    }
	    return list;
	}
	
	@SuppressWarnings("unchecked")
	protected <K> List<B> loadByPks(K... keys){
	    if(null == keys){
	    	return Collections.emptyList();
	    }
		return loadByPks(Arrays.asList(keys));
	}

	@Override
    public boolean existsByPrimaryKey(B bean)throws RuntimeDaoException{
    	return null == bean ? false : existsPrimaryKey(bean.primaryValues());
    }
    @Override
    public B checkDuplicate(B bean)throws RuntimeDaoException,ObjectRetrievalException{
        if(existsByPrimaryKey(bean)){
            throw new ObjectRetrievalException("Duplicate entry ("+ bean.primaryValues() +") for key 'PRIMARY'");
        }
        return bean;   
    }
    @Override
    public final boolean existsPrimaryKey(Object ...keys)throws RuntimeDaoException{
    	// {{check parameters
    	if(metaData.primaryKeyNames.length == 0){
    		throw new UnsupportedOperationException();
    	}
    	String[] pkNames = metaData.primaryKeyNames;
    	int[] pkIds = metaData.primaryKeyIds;
    	if(null == keys || hasNull(keys)){
    		return false;
    	}
    	checkArgument(keys.length == pkNames.length,
    			"INVALID ARGUMENT NUM, %s required",pkNames.length);
    	for(int i=0; i<pkNames.length; ++i){
    		Object key = keys[i];
    		Class<?> type = metaData.columnTypeOf(pkIds[i]);
    		checkArgument(type.isInstance(key),
    				"INVALID type for pk %s,%s required",pkNames[i],type.getName());
    	}
    	// }}check parameters
    	return doExistsPrimaryKey(keys);
    }
    protected boolean doExistsPrimaryKey(Object ...keys)throws RuntimeDaoException{
        String[] pkNames = metaData.primaryKeyNames;
        int[] pkIds = metaData.primaryKeyIds;
        Connection c = null;
        PreparedStatement ps = null;
        try{
            c = this.getConnection();
            String col = 1 == metaData.primaryKeyCount ? metaData.primaryKeyNames[0] : "1";
            StringBuilder builder = new StringBuilder("SELECT COUNT(" + col + ") AS MCOUNT FROM "  + metaData.tablename + " WHERE ");
            builder.append(Joiner.on(" AND ").join(Lists.transform(Arrays.asList(pkNames),columnExpFun)));
            ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), false, debug, 
                    "ExistsPrimaryKey", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);

			for(int i = 0; i<pkNames.length; ++i){
				 setPreparedStatement(ps, i+1, keys[i], pkIds[i]);
			}
            return 1 == getManager().runPreparedStatementForValue(Number.class, ps).intValue();
        }catch(SQLException e){
            throw new RuntimeDaoException(new ObjectRetrievalException(e));
        }finally{
            this.getManager().close(ps);
            this.freeConnection(c);
        }
    }
    protected <T>T checkDuplicateByPk(T primaryKeyValue)throws ObjectRetrievalException{
    	if(metaData.primaryKeyNames.length != 1){
    		throw new UnsupportedOperationException();
    	}
	    if(existsPrimaryKey(primaryKeyValue)){
	        throw new ObjectRetrievalException("Duplicate entry '"+ primaryKeyValue +"' for key 'PRIMARY'");
	    }
	    return primaryKeyValue;
	}

	@SuppressWarnings("unchecked")
    @Override
    public B[] loadByWhere(String where)throws RuntimeDaoException{
	    return this.loadByJoinWhereAsList(null,where, null, 1, -1).toArray((B[])Array.newInstance(metaData.beanType,0));
    }
	
    @Override
    public int loadByWhere(String where, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByJoinWhereForAction(null,where, null,null, 1,-1,action);
    }

    @SuppressWarnings("unchecked")
    @Override
    public B[] loadByWhere(String where, int[] fieldList)throws RuntimeDaoException{
        return this.loadByJoinWhereAsList(null,where, fieldList, 1, -1).toArray((B[])Array.newInstance(metaData.beanType,0));
    }

    @Override
    public int loadByWhere(String where, int[] fieldList, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByJoinWhereForAction(null,where, null, fieldList, 1, -1, action);
    }

    @SuppressWarnings("unchecked")
    @Override
    public B[] loadByWhere(String where, int[] fieldList, int startRow, int numRows)throws RuntimeDaoException{
        return this.loadByJoinWhereAsList(null,where, fieldList, startRow, numRows).toArray((B[])Array.newInstance(metaData.beanType,0));
    }
    
    @Override
    public int loadByWhere(String where, int[] fieldList, int startRow, int numRows,
            TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByJoinWhereForAction(null,where, null, fieldList, startRow, numRows, action);
    }

    @Override
    public List<B> loadByWhereAsList(String where)throws RuntimeDaoException{
        return this.loadByJoinWhereAsList(null,where, null, 1, -1);
    }
    
    @Override
    public List<B> loadByJoinWhereAsList(String join,String where)throws RuntimeDaoException{
        return this.loadByJoinWhereAsList(join,where, null, 1, -1);
    }

    @Override
    public List<B> loadByWhereAsList(String where, int[] fieldList)throws RuntimeDaoException{
        return this.loadByJoinWhereAsList(null,where, fieldList, 1, -1);
    }

    @Override
    public List<B> loadByJoinWhereAsList(String join ,String where, Object[] argList, int[] fieldList)throws RuntimeDaoException{
    	ListAction<B> action = new ListAction<>();
    	loadByJoinWhereForAction(join,where, argList,fieldList, 1, -1,action);
		return action.getList();
    }

    @Override
    public List<B> loadByWhereAsList(String where, int[] fieldList, int startRow, int numRows)throws RuntimeDaoException{
        return loadByJoinWhereAsList(null,where, fieldList, startRow, numRows);
    }

    @Override
    public List<B> loadByJoinWhereAsList(String join ,String where, int[] fieldList, int startRow, int numRows)throws RuntimeDaoException{
        ListAction<B> action = new ListAction<>();
        loadByJoinWhereForAction(join,where, null, fieldList, startRow, numRows, action);
        return action.getList();
    }
    @Override
    public <T>List<T> loadByJoinWhereAsList(String join ,String where,Object[] argList, int[] fieldList, int startRow, int numRows,Function<B, T>transformer) throws RuntimeDaoException{
    	ListAction<B> action = new ListAction<>();
        loadByJoinWhereForAction(join, where, argList, fieldList, startRow, numRows, action);
        return action.getList(transformer);
    }

    @Override
    public int loadByWhereForAction(String where, Object[] argList, int[] fieldList, int startRow,int numRows, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByJoinWhereForAction(null,where, argList, fieldList, startRow, numRows, action);
    }
    
    @Override
    public int loadByJoinWhereForAction(String join,String where, Object[] argList, int[] fieldList, int startRow,
            int numRows, TableManager.Action<B> action)throws RuntimeDaoException{
        String sql = createSelectSql(fieldList, join, where);
        return this.loadBySqlForAction(sql, argList, fieldList, startRow, numRows, action);
    }

    @Override
    public B[] loadUsingTemplate(B bean)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, 1, -1, SEARCH_EXACT);
    }

    @Override
    public int loadUsingTemplate(B bean, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, null, 1, -1, SEARCH_EXACT, action);
    }

    @Override
    public B[] loadUsingTemplate(B bean, int startRow, int numRows)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, startRow, numRows, SEARCH_EXACT);
    }

    @Override
    public int loadUsingTemplate(B bean, int startRow, int numRows,
            TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, null, startRow, numRows,SEARCH_EXACT, action);
    }

    @SuppressWarnings("unchecked")
    @Override
    public B[] loadUsingTemplate(B bean, int startRow, int numRows, int searchType)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(bean, startRow, numRows, searchType).toArray((B[])Array.newInstance(metaData.beanType,0));
    }

    @Override
    public List<B> loadUsingTemplateAsList(B bean)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(bean, 1, -1, SEARCH_EXACT);
    }

    @Override
    public List<B> loadUsingTemplateAsList(B bean, int startRow, int numRows)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(bean, startRow, numRows, SEARCH_EXACT);
    }

    @Override
    public List<B> loadUsingTemplateAsList(B bean, int startRow, int numRows, int searchType)throws RuntimeDaoException{
    	ListAction<B> action = new ListAction<>();
        loadUsingTemplate(bean,null,startRow,numRows,searchType, action);
        return action.getList();
    }
    //18
    @Override
    public B loadUniqueUsingTemplate(B bean)
    {
    	try {
    		return loadUniqueUsingTemplateChecked(bean);	
		} catch (ObjectRetrievalException e) {
			return null;
		}
     }
    //18-1
    @Override
    public B loadUniqueUsingTemplateChecked(B bean) throws ObjectRetrievalException
    {
        List<B> beans = runWithNoPage(new Callable<List<B>>() {
            @Override
            public List<B> call() throws Exception {
                return loadUsingTemplateAsList(bean);
            }
        });
        switch(beans.size()){
        case 0:
            throw new ObjectRetrievalException("Not found element !!");
        case 1:
            return beans.get(0);
        default:
            throw new RuntimeDaoException(new ObjectRetrievalException("More than one element !!"));
        }
    }
    
    @Override
    public int loadUsingTemplate(B bean, int[] fieldList, int startRow, int numRows,int searchType, Action<B> action)
    {
        StringBuilder sqlWhere = new StringBuilder("");
        String sql=createSelectSql(fieldList,
        		null, this.fillWhere(sqlWhere, bean, searchType) > 0 ? " WHERE "+ sqlWhere.toString() : null);
        PreparedStatement ps = null;
        Connection c = null;
        try {
            c = this.getConnection();
            AtomicLong count = new AtomicLong(-1L);
            Manager.setLocalfillPreparedStatement((BaseRow)bean, searchType, false);
            String wrapped = getManager().rebuildSelectSql(c,sql,null, startRow, numRows, count, debug);
            // 执行count语句返回0,就不必再继续执行SQL查询
            if(0 == count.get()){
            	return 0;
            }
//            PageQueryImplType pageQueryImplType = getManager().getPageQueryImplType(connection);
//            String wrapped = pageQueryImplType.wrap(sql, startRow, numRows);
            sql = firstNonNull(wrapped, sql);
            ps = getManager().getStatementCache().prepareStatement(c, sql, false, debug, 
                    "loadUsingTemplate", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            this.fillPreparedStatement(ps, bean, searchType,false, true, null, 0);
            if(null != wrapped){
                return this.loadByPreparedStatement(ps, fieldList, 1, -1, action);
            }else{
                return this.loadByPreparedStatement(ps, fieldList, startRow, numRows, action);
            }
        } catch (DaoException e) {
            throw new RuntimeDaoException(e);
        }catch (SQLException e) {
            throw new RuntimeDaoException(new DataAccessException(e));
        } finally {
            this.getManager().close(ps);
            this.freeConnection(c);
            Manager.removeLocalfillPreparedStatement();
        }
    }
    /**
     * @param <F> bean type of foreign table 
     * @param left
     * @param columnsMap
     * @param startRow
     * @param numRows
     * @return list of B or empty list
     */
    private <F extends BaseBean>
	List<B> loadByForeignKeyAsList(F left,Map<Integer,Integer>columnsMap,int startRow, int numRows)
	{
	    if(null == left){
	        return Collections.emptyList();
	    }
		checkArgument(columnsMap != null && !columnsMap.isEmpty(),"columnsMap is null or empty");
	    B bean = createBean().copy(left, columnsMap);
	    return loadUsingTemplateAsList(bean,startRow,numRows);
	}
    /**
     * @param fkName
     * @param left
     * @param startRow
     * @param numRows
     * @param <F> bean type of foreign table
     * @return list of B or empty list
     */
    protected <F extends BaseBean>
	List<B> loadByForeignKeyAsList(String fkName,F left,int startRow, int numRows)
	{
	    ImmutableBiMap<Integer, Integer> columnsMap = metaData.foreignKeyIdMapOf(fkName).inverse();
	    return loadByForeignKeyAsList(left,columnsMap,startRow,numRows);
	}

    @Override
    public void foreachByWhere(DoEach<B> each, boolean stopOnError,String where)throws RuntimeDaoException{
        foreachByJoinWhere(each,stopOnError,null,where);
    }
    public void foreachByJoinWhere(DoEach<B> each, boolean stopOnError,String join,String where)throws RuntimeDaoException{
        checkArgument(each != null, "action is null");
        List<B> beans = loadByJoinWhereAsList(join,where, null, null);
        for(B bean : beans){
            try {
                if(each.doEach(bean)){
                    delete(bean);
                }
            } catch (Exception e) {
                if(stopOnError){
                    return;
                }
                continue;
            }
        }
    }
    @Override
    public void foreach(DoEach<B> each, boolean stopOnError)throws RuntimeDaoException{
    	foreachByWhere(each,stopOnError,null);
    }
	@Override
    public B save(B bean)throws RuntimeDaoException{
        if(null != bean){
            if (bean.isNew()) {
                this.insert(bean);
            } else {
                this.update(bean);
            }
        }
        return bean;
    }
	
	@Override
	public B addIfAbsent(final B bean)throws RuntimeDaoException{
		try {
			if(bean == null){
				return bean;
			}
			// force to set to new 
			bean.setNew(true);
			return insert(bean);
		} catch (RuntimeDaoException e) {
			if (stripSQLException(e) instanceof SQLIntegrityConstraintViolationException) {
				B exists = loadByPrimaryKey(bean);
				if(exists != null){
					// duplicated primary key 
					// return this exists row
					return exists;
				}
				// throw while other column duplicated 
			}					
			throw e;
		}	
    }
	
    @Override
    public B[] save(B[] beans)throws RuntimeDaoException{
        if(null != beans){
            for (B bean : beans) 
            {
                this.save(bean);
            }
        }
        return beans;
    }

    @Override
    public <C extends Collection<B>> C save(C beans)throws RuntimeDaoException{
        if(null != beans){
            for (B bean : beans) 
            {
                this.save(bean);
            }
        }
        return beans;
    }
    
    @Override
    public <C extends Collection<B>> C saveAsTransaction(final C beans)throws RuntimeDaoException{
        return this.runAsTransaction(new Callable<C>(){
            @Override
            public C call() throws Exception {
                return save(beans);
            }});
    }

    @Override
    public B[] saveAsTransaction(final B[] beans)throws RuntimeDaoException{
        return this.runAsTransaction(new Callable<B[]>(){
            @Override
            public B[] call() throws Exception {
                return save(beans);
            }});
    }

    /**
     * JOIN SQL 语句安全性(防注入攻击)检查
     * @param join
     * @return join
     */
    static String checkJoin(String join){
        join = MoreObjects.firstNonNull(join, "").trim();
        if(!join.isEmpty()){
        	checkArgument(join.toUpperCase().matches("^(JOIN|LEFT|RIGHT) +.*"),
    				"JOIN expression must start with 'JOIN|LEFT|RIGHT'(case insensitive)");
        }
        return join;
    }
    @SuppressWarnings("unchecked")
	@Override
    public <T> List<T> loadColumnAsList(String column,boolean distinct,String where,int startRow,int numRows)throws RuntimeDaoException{
        int columnId = metaData.columnIDOf(column);
        checkArgument(columnId>=0,"INVALID column name %s",column);
        String sql = String.format("SELECT %s " + metaData.columnNameOf(columnId) + " FROM %s %s",
                distinct ? "DISTINCT" : "",
                metaData.tablename,
                MoreObjects.firstNonNull(where, ""));
        return runSqlAsList((Class<T>)metaData.columnTypes.get(columnId), sql);
    }

    /**
     * generate SQL query(SELECT) statement,such as: 'SELECT id,name from mytable WHERE id=1'
     * @param fieldList
     * @param join JOIN statement
     * @param where where condition expression statement that start with 'WHERE',or {@code null},or empty string
     * @return SQL statement string
     * @throws IllegalArgumentException where condition expression don't start with 'WHERE'
     */
    private String createSelectSql(int[] fieldList, String join, String where){
        StringBuffer sql = new StringBuffer(128);
        
        if(null == fieldList || 0 == fieldList.length) {
            sql.append("SELECT ").append(metaData.columnFullFields);
        } else{
            /** filter invalid column id*/ 
            Iterable<Integer> validIdx = Iterables.filter(Ints.asList(fieldList), idx->null != idx && idx >=0 && idx < metaData.columnCount);
            sql.append("SELECT ").append(Joiner.on(',').join(Iterables.transform(validIdx,metaData.columnFullFieldList::get)));
        }
        sql.append(" FROM " + metaData.tablename + " ");
        join = checkJoin(join);
        if(!join.isEmpty()){
            sql.append(join + " ");
        }
        sql.append(MoreObjects.firstNonNull(where, ""));
        return sql.toString();
    }

    //2.2

    @Override
    public int delete(B bean){
    	if(metaData.primaryKeyNames.length==0){
    		throw new UnsupportedOperationException();
    	}
        if(hasNullPk(bean)){
        	return 0;
        }
        Connection c = null;
        PreparedStatement ps = null;
        String [] pks = metaData.primaryKeyNames;
        int[] pkIds = metaData.primaryKeyIds;
        try
        {
            // listener callback
            getListenerContainer().beforeDelete(bean);
            c = this.getConnection();
            StringBuilder builder = new StringBuilder("DELETE  FROM " + metaData.tablename +" WHERE ");
            builder.append(Joiner.on(" AND ").join(Lists.transform(Arrays.asList(pks),columnExpFun)));
            ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), false, debug, 
                    "deleteByPrimaryKey", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            for(int i = 0; i<pkIds.length; ++i){
            	 setPreparedStatementWithBean(ps, i+1, bean, pkIds[i]);
            }
            int rows=ps.executeUpdate();
            if(rows>0){
                // listener callback
                getListenerContainer().afterDelete(bean);
            }
            return rows;
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            // listener callback
            getListenerContainer().done();
            getManager().close(ps);
            freeConnection(c);
        }
    }

    protected <K> int deleteByPks(Collection<K> keys){
		checkState(metaData.primaryKeyCount == 1,"UNSUPPORTED OPERATION");
	    int count = 0;
	    if(null != keys){
	        for(K key :keys){
	            count += deleteByPrimaryKey(key);
	        }
	    }
	    return count;
	}

	@SuppressWarnings("unchecked")
	protected <K> int deleteByPks(K... keys){
	    if(null == keys){
	    	return 0;
	    }
	    return deleteByPks(Arrays.asList(keys));
	}

	@Override
	public int deleteByPrimaryKey(Object ...keys)throws RuntimeDaoException{
	    if(hasNull(keys)){
	    	return 0;
	    }
		checkArgument(keys.length == metaData.primaryKeyCount,"argument number mismatch with primary key number");
	    return delete(createBean(keys));
	}

	//2.4 

	@SuppressWarnings("unchecked")
	@Override
	public int delete(B... beans)throws RuntimeDaoException{
	    int count = 0;
	    if(null != beans){
	        for(B bean :beans){
	            count += delete(bean);
	        }
	    }
	    return count;
	}

	//2.5  

	@Override
	public int delete(Collection<B> beans)throws RuntimeDaoException{
	    int count = 0;
	    if(null != beans){
	        for(B bean :beans){
	            count += delete(bean);
	        }
	    }
	    return count;
	}

	@Override
	public boolean setValueIfNonNull(B bean,String column, Object value){
        B old;
        if(null != bean && null != (old = loadByPrimaryKey(bean))){
            return bean.setValueIf(null != old.getValue(column), column, value);
        }
        return false;
    }
	
	@Override
    public boolean setValueIfNonEqual(B bean,String column, Object value){
	    B old;
	    if(null != bean && null != (old = loadByPrimaryKey(bean))){
            return bean.setValueIf(!Objects.equals(old.getValue(column), value), column, value);
	    }
	    return false;
	}
	
	/**
     * Retrieves the T object referenced by fkName.<br>
	 * @param fkName foreign key name.<br>
	 * @param bean the B object to use
     * @param <T>
     * @return the associated T bean or {@code null} if {@code bean}  is {@code null}
     * @throws RuntimeDaoException
     */
	protected <T extends BaseBean> T getReferencedBean(String fkName, B bean) throws RuntimeDaoException{
    	if(null == bean){
    		return null;
    	}
    	ForeignKeyMetaData foreignkey = checkNotNull(metaData.foreignKeys.get(fkName),
    			"INVALID fkName %s for table %s", fkName, metaData.tablename);
		BaseTableManager<T> foreignManager = baseManagerOf(foreignkey.foreignTable);
    	T t = foreignManager.createBean().copy(bean, metaData.foreignKeyIdMapOf(fkName));
    	return foreignManager.loadByPrimaryKey(t);
    }
    
    /**
     * Associates the B object to the T object by fkName field.<br>
     * @param fkName see also {@link #getReferencedBean(String, BaseBean)}
     * @param bean the B object to use
     * @param beanToSet the T object to associate to the B bean
     * @param <T> see also {@link #getReferencedBean(String, BaseBean)}
     * @return always beanToSet saved
     * @throws RuntimeDaoException
     */
	protected <T extends BaseBean> T setReferencedBean(String fkName, B bean, T beanToSet) throws RuntimeDaoException{
        if(null != bean){
        	ForeignKeyMetaData foreignkey = checkNotNull(metaData.foreignKeys.get(fkName),
        			"INVALID fkName %s for table %s", fkName, metaData.tablename);
    		TableManager<BaseBean> foreignManager = managerOf(foreignkey.foreignTable);
    		foreignManager.save(beanToSet);
           	bean.copy(beanToSet, metaData.foreignKeyIdMapOf(fkName).inverse());
        }
        return beanToSet;
    }
    /**
     * Retrieves imported T objects by fkIndex.<br>
     * @param <T>
     * @param fkName foreign key name.<br>
     * @param bean the B object to use
     * @return the associated T beans or {@code null} if {@code bean} is {@code null}
     * @throws RuntimeDaoException
     */
    @SuppressWarnings("unchecked")
	protected <T extends BaseBean> T[] getImportedBeans(String fkName,B bean) throws RuntimeDaoException{
    	ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
    	RowMetaData foreignMetaData = RowMetaData.getMetaData(foreignkey.foreignTable);
    	return getImportedBeansAsList(fkName,bean).toArray((T[]) Array.newInstance(foreignMetaData.beanType, 0));
    }

    /**
     * Retrieves imported T objects by fkIndex.<br>
     * @param <T>
     * @param fkName foreign key name.<br>
     * @param bean the B object to use
     * @return the associated T beans or {@code null} if {@code bean} is {@code null}
     * @throws RuntimeDaoException
     */
	protected <T extends BaseBean> List<T> getImportedBeansAsList(String fkName, B bean,int startRow, int numRows)throws RuntimeDaoException{
    	ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
		TableManager<T> foreignManager = managerOf(foreignkey.ownerTable);
		return baseManagerOf(foreignManager).loadByForeignKeyAsList(fkName,bean, startRow, numRows);
    }

    /**
     * Retrieves imported T objects by ikIndex.<br>
     * @param <T> see also {@link #getImportedBeans(String, BaseBean)}
     * @param fkName foreign key name.see also {@link #getImportedBeans(String, BaseBean)}
     * @param bean the B object to use
     * @return the associated T beans or {@code null} if {@code bean} is {@code null}
     * @throws RuntimeDaoException
     */
	protected <T extends BaseBean> List<T> getImportedBeansAsList(String fkName, B bean)throws RuntimeDaoException{
		return getImportedBeansAsList(fkName,bean, 1, -1);
    }
    protected <T extends BaseBean> List<T> getImportedBeansAsList(String fkName,Object...keys) throws RuntimeDaoException{		
    	return getImportedBeansAsList(fkName,createBean(keys),1,-1);
    }
	@SuppressWarnings("unchecked")
	protected <T extends BaseBean> T[] getImportedBeans(String fkName,Object...keys) throws RuntimeDaoException{		
    	ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
    	RowMetaData foreignMetaData = RowMetaData.getMetaData(foreignkey.foreignTable);
    	return getImportedBeansAsList(fkName,keys).toArray((T[]) Array.newInstance(foreignMetaData.beanType, 0));
    }
    /**
     * Set the importedBeans associates to the bean by fkIndex<br>
     * @param fkName foreign key name. see also {@link #getImportedBeans(String, BaseBean)}
     * @param bean the bean object to use
     * @param importedBeans the T object to associate to bean
     * 
     * @param <T> see also {@link #getImportedBeans(String, BaseBean)}
     * @return importedBeans always
     * @throws RuntimeDaoException
     */
	protected <T extends BaseBean, C extends Collection<T>> C setImportedBeans(String fkName, B bean,
            C importedBeans)throws RuntimeDaoException{
        if(null != importedBeans){
        	ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
    		BaseTableManager<BaseBean> foreignManager = baseManagerOf(foreignkey.ownerTable);
        	for( T importBean : importedBeans ){
        		foreignManager.setReferencedBean(fkName , importBean,bean);
            }
        }
        return importedBeans;
    }
    
    
    /**
     * Set the importedBeans associates to the bean by {@code ikIndex}<br>
     * @param fkName foreign key name.see also {@link #getImportedBeans(String, BaseBean)}
     * @param bean the bean object to use
     * @param importedBeans the T object to associate to bean
     * 
     * @param <T> see also {@link #getImportedBeans(String, BaseBean)}
     * @return importedBeans always
     * @throws RuntimeDaoException
     */
    protected <T extends BaseBean> T[] setImportedBeans(String fkName, B bean, T[] importedBeans) throws RuntimeDaoException{
    	if(null != importedBeans){
    		setImportedBeans(fkName, bean, Arrays.asList(importedBeans));
    	}
    	return importedBeans;
    }
    /**
     * delete all imported beans by fkName 
     * @param fkName foreign key name
     * @param bean 
     * @return deleted row count or 0 if bean is null 
     */
	protected int deleteImportedBeans(String fkName,B bean) throws RuntimeDaoException{
    	if(bean == null){
    		return 0;
    	}
    	ForeignKeyMetaData foreignkey = metaData.getImportedKey(fkName);
    	RowMetaData ownerTable = RowMetaData.getMetaData(foreignkey.ownerTable);
		BaseTableManager<BaseBean> foreignManager = baseManagerOf(foreignkey.ownerTable);
		BaseBean tmpl = foreignManager.createBean().copy(bean,ownerTable.foreignKeyIdMapOf(fkName));
		return foreignManager.deleteUsingTemplate(tmpl);
    }
    protected int deleteImportedBeans(String fkName, Map<Integer, Object> idValueMap) throws RuntimeDaoException{
    	B bean = createBean().copyNoFilter(idValueMap);
    	return deleteImportedBeans(fkName,bean);
    }
    
    protected int deleteImportedBeans(String fkName,Object...keys) throws RuntimeDaoException{
    	return deleteImportedBeans(fkName,createBean(keys));
    }
    private Map<Integer, Object> makeIndexValueMap(String indexName,Object ...indexValues){
    	IndexMetaData indexMetaData = metaData.indices.get(indexName);
    	checkArgument(null != indexMetaData,"INVALID index name %s",indexName);
    	checkArgument(null != indexValues && indexValues.length == indexMetaData.columns.size(),"INVALID index value");
    	ImmutableMap.Builder<Integer,Object> builder = ImmutableMap.<Integer,Object>builder();
    	for(int i=0;i<indexValues.length;++i){
    		Object value = indexValues[i];
    		if(null != value){
    			String column = indexMetaData.columns.get(i);
    			builder.put(metaData.columnIDOf(column), value);
    		}
    	}
    	return builder.build();
    }
 
    //_____________________________________________________________________
    //
    // USING INDICES
    //_____________________________________________________________________    
    /**
     * Retrieves a array of B bean using the index specified by keyIndex.
     * @param indexName name of index
     * @param keys key values of index
     * @return B array
     * @throws RuntimeDaoException
     * @see #loadByIndexAsList(String, Object...)
     */
    @SuppressWarnings("unchecked")
    protected B[] loadByIndex(String indexName,Object ...keys)throws RuntimeDaoException{
        return this.loadByIndexAsList(indexName,keys).toArray((B[])Array.newInstance(metaData.beanType,0));
    }
    /**
     * Retrieves a list of B bean using the index specified by indexName.
     * @param indexName name of index<br>
     * @param indexValues key values of index
     * @return a list of B bean
     * @throws RuntimeDaoException
     */
    protected List<B> loadByIndexAsList(String indexName,Object ...indexValues)throws RuntimeDaoException{
    	Map<Integer, Object> map = makeIndexValueMap(indexName,indexValues);
    	return loadUsingTemplateAsList(map);
    }
    protected final B loadUniqueByIndex(String indexName,Object ...indexValues)throws RuntimeDaoException{
        return doLoadUniqueByIndex(indexName, indexValues);
    }
    protected B doLoadUniqueByIndex(String indexName,Object ...indexValues)throws RuntimeDaoException{
        if(hasNull(indexValues)){
            return null;
        }
    	Map<Integer, Object> keys = makeIndexValueMap(indexName,indexValues);
    	B bean = createBean().copyNoFilter(keys);
        return loadUniqueUsingTemplate(bean);
    }
    protected final B loadUniqueByIndexChecked(String indexName,Object ...indexValues)throws ObjectRetrievalException{
		if(hasNull(indexValues)){
			throw new ObjectRetrievalException(new NullPointerException("index keys must not be null"));
		}
    	return doLoadUniqueByIndexChecked(indexName, indexValues);
    }
    protected B doLoadUniqueByIndexChecked(String indexName,Object ...indexValues)throws ObjectRetrievalException{
    	Map<Integer, Object> keys = makeIndexValueMap(indexName,indexValues);
    	B bean = createBean().copyNoFilter(keys);
        return loadUniqueUsingTemplateChecked(bean);
    }
    /**
     * Deletes rows using key.
     * @param indexName name of index
     * @param indexValues key values of index
     * @return the number of deleted objects
     * @throws RuntimeDaoException
     */
    protected int deleteByIndex(String indexName,Object ...indexValues)throws RuntimeDaoException{
    	Map<Integer, Object> map = makeIndexValueMap(indexName,indexValues);
    	return deleteUsingTemplate(map);
    }

    @SuppressWarnings("unchecked")
	protected <T extends BaseBean> List<B> loadViaJunctionAsList(String junctionTable,T linked, int startRow, int numRows){
    	BaseTableManager<BaseBean> manager = baseManagerOf(junctionTable);
    	return manager.loadViaJunctionTableAsList((Class<B>)metaData.beanType, linked, startRow, numRows);
    }
    protected <T extends BaseBean> List<B> loadViaJunctionAsList(String junctionTable,T linked){
    	return loadViaJunctionAsList(junctionTable,linked,1,-1);
    }
    private List<B> loadUsingTemplateAsList(Map<Integer, Object>keys){
    	B bean = createBean().copyNoFilter(keys);
    	return loadUsingTemplateAsList(bean);
    }
    
    private int deleteUsingTemplate(Map<Integer, Object>keys){
    	B bean = createBean().copyNoFilter(keys);
    	return deleteUsingTemplate(bean);
    }
    protected <T>List<B> loadByIndexForIndices(String indexName,Collection<T> indexs)
    {
    	IndexMetaData indexMetaData = metaData.indices.get(indexName);
    	checkArgument(null != indexMetaData,"INVALID index name %s",indexName);
    	checkArgument(1 == indexMetaData.columns.size(),"column count of  index must be 1");
    	checkArgument(indexMetaData.unique,"index must be unique");
    	if(null == indexs ){
            return Collections.emptyList();
        }
        List<B> list = new ArrayList<B>(indexs.size());
        for(T key: indexs){
            list.add(loadUniqueByIndex(indexName,key));
        }
        return list;
    }
    @SuppressWarnings("unchecked")
	protected <T>List<B> loadByIndexForIndices(String indexName,T... indexs){
    	if(indexs == null || indexs.length == 0){
    		return Collections.emptyList();
    	}
    	return loadByIndexForIndices(indexName,Arrays.asList(indexs));
    }
    protected <T>int deleteByIndexForIndices(String indexName,Collection<T> indexs)
    {
    	IndexMetaData indexMetaData = metaData.indices.get(indexName);
    	checkArgument(null != indexMetaData,"INVALID index name %s",indexName);
    	checkArgument(1 == indexMetaData.columns.size(),"column count of  index must be 1");

    	if(null == indexs ){
            return 0;
        }
        int count = 0;
        for(T key: indexs){
        	if(key != null){
        		count += deleteByIndex(indexName, key);
        	}
        }
        return count;
    }
    @SuppressWarnings("unchecked")
	protected <T>int deleteByIndexForIndices(String indexName,T... indexs){
    	if(indexs == null || indexs.length == 0){
    		return 0;
    	}
    	return deleteByIndexForIndices(indexName,Arrays.asList(indexs));
    }
    private boolean checkPkValid(B bean)
	{
		if(metaData.primaryKeyNames.length ==0){
			return false;
		}
    	for(String name:metaData.primaryKeyNames){
    		if(! (bean.isInitialized(name) && null != bean.getValue(name))){
    			return false;	
    		}
    	}	
    	return true;
	}
    //21

    @Override
    public int deleteUsingTemplate(B bean)
    {
    	if(bean == null || !bean.beModified()){
    		return 0;
    	}
        if(checkPkValid(bean)){
            return this.deleteByPrimaryKey(bean);
        }
        if( !getListenerContainer().isEmpty()){
            DeleteBeanAction action=new DeleteBeanAction(); 
            this.loadUsingTemplate(bean,action);
            return action.getCount();
        }
        Connection c = null;
        PreparedStatement ps = null;
        StringBuilder builder = new StringBuilder("DELETE FROM " + metaData.tablename + " ");
        StringBuilder sqlWhere = new StringBuilder("");

        try
        {
            fillWhere(sqlWhere, bean, SEARCH_EXACT);
            builder.append(" WHERE ").append(sqlWhere);
            c = this.getConnection();
            ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), false, debug, 
                    "deleteUsingTemplate", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            this.fillPreparedStatement(ps, bean, SEARCH_EXACT, false, true, null, 0);

            return ps.executeUpdate();
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            this.getManager().close(ps);
            this.freeConnection(c);
        }
    }
    
    //11

    @Override
    public int deleteByWhere(String where)
    {
        if( !getListenerContainer().isEmpty()){
            final DeleteBeanAction action = new DeleteBeanAction(); 
            this.loadByWhere(where,action);
            return action.getCount();
        }
        Connection c = null;
        PreparedStatement ps = null;

        try
        {
            c = this.getConnection();
            StringBuilder builder = new StringBuilder("DELETE " + metaData.tablename + " FROM " + metaData.tablename + " " + where);
            ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), true, debug, "deleteByWhere");
            return ps.executeUpdate();
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            this.getManager().close(ps);
            this.freeConnection(c);
        }
    }
    private void setPreparedStatement(PreparedStatement ps,int pos,Object value,int columnId)
    		throws SQLException {
        getManager().fillPreparedStatement(ps, pos, value, metaData.sqlTypes[columnId], metaData.columnTypeNames.get(columnId));
    }
    private void setPreparedStatementWithBean(PreparedStatement ps,int pos,B bean,int columnId)
    		throws SQLException {
    	setPreparedStatement(ps,pos,bean.getJdbcValue(columnId),columnId);
    }
    
    //3.9 SYNC SAVE 
    /**
     * Save the B bean and referenced beans and imported beans into the database.
     *
     * @param bean the B bean to be saved
     * @param referenceBeans referenced beans beans<br>
     * @param importedBeans imported beans <br>
     * @return the inserted or updated B bean
      * @throws RuntimeDaoException
     */
	private B save(B bean,Map<String, BaseBean> referenceBeans,Map<String, Collection<BaseBean>> importedBeans) throws RuntimeDaoException{
	    if(null == bean){
	        return null;
	    }
	    referenceBeans = MoreObjects.firstNonNull(referenceBeans, Collections.<String, BaseBean>emptyMap());
	    importedBeans = MoreObjects.firstNonNull(importedBeans, Collections.<String, Collection<BaseBean>>emptyMap());
	    for(Entry<String, BaseBean> entry:referenceBeans.entrySet()){
	    	BaseBean beanToSet = entry.getValue();
	    	if(beanToSet != null){
	    		setReferencedBean(entry.getKey(), bean, beanToSet);
	    	}
	    }
	    bean = this.save( bean );
	    for(Entry<String, Collection<BaseBean>> entry:importedBeans.entrySet()){
	    	String ikName = entry.getKey();
	    	Collection<BaseBean> importeds = entry.getValue(); 
	    	if(null != importeds){
	    		setImportedBeans(ikName, bean, importeds);
	    		TableManager<BaseBean> impManager = managerOf(metaData.getImportedKey(ikName).ownerTable);
	    		impManager.save(importeds);
	    	}
	    }
		return bean;
	}
    
    @SuppressWarnings("unchecked")
	protected B saveFully(B bean,Object[] args){
    	HashMap<String,BaseBean> referenced = Maps.newHashMap();
    	int index = 0;
    	for(String key : metaData.foreignKeys.keySet()){
    		referenced.put(key, (BaseBean) args[index++]);
    	}
    	HashMap<String,Collection<BaseBean>> imported = Maps.newHashMap();
    	for(ForeignKeyMetaData key : metaData.getImportedKeys()){
    		Object obj = args[index++];
    		if(obj == null){
    			imported.put(key.name, null);
    		}else if(obj.getClass().isArray()){
    			checkArgument(BaseBean.class.isAssignableFrom(obj.getClass().getComponentType()),
    					"INVALID COMPONENT TYPE FOR %s",key.name);
    			imported.put(key.name,Arrays.asList((BaseBean[])obj));
    		}else if(obj instanceof Collection){
    			imported.put(key.name,(Collection<BaseBean>)obj);
    		}else{
    			throw new IllegalArgumentException(logString("INVALID TYPE %s FOR %s",obj.getClass(),key.name));
    		}    		
    	}
    	return save(bean,referenced,imported);
    }
    
    protected B saveFullyAsTransaction(final B bean, final Object[] args){
    	return this.runAsTransaction(new Callable<B>(){
            @Override
            public B call() throws Exception {
                return saveFully(bean , args );
            }});
    }
    //_____________________________________________________________________
    //
    // COUNT
    //_____________________________________________________________________
    //25

    @Override
    public int countWhere(String where)
    {
        return rowCountWhere(where);
    }
    @Override
	public int rowCountWhere(String where, Object... argList)
    {
        String col = 1 == metaData.primaryKeyCount  ? metaData.primaryKeyNames[0] : "1";
    	String sql = new StringBuffer("SELECT * FROM " + metaData.tablename + " ")
    			.append(MoreObjects.firstNonNull(where, "")).toString();
        String countSql = ParserSupport.countSql(sql, col);
    	return runSqlForValue(Long.class, countSql, argList).intValue();
    }

    //20
    /**
     * count the number of elements of a specific B bean given the search type
     *
     * @param bean the B template to look for
     * @param searchType exact ?  like ? starting like ?
     * @return the number of rows returned
     */
    @Override
    public int countUsingTemplate(B bean, int searchType)
    {
        Connection c = null;
        PreparedStatement ps = null;
        String col = 1 == metaData.primaryKeyCount  ? metaData.primaryKeyNames[0] : "1";
        StringBuilder builder = new StringBuilder("SELECT COUNT(" + col + ") AS MCOUNT FROM " + metaData.tablename);
        StringBuilder sqlWhere = new StringBuilder("");

        try
        {
            if (this.fillWhere(sqlWhere, bean, SEARCH_EXACT) > 0) {
                builder.append(" WHERE ").append(sqlWhere);
            } else {
            	log(debug,"The bean to look is not initialized... counting all...");
            }

            c = this.getConnection();
            ps = getManager().getStatementCache().prepareStatement(c, builder.toString(), false, debug, 
                    "countUsingTemplate", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            this.fillPreparedStatement(ps, bean, searchType,false, true, null, 0);
            return getManager().runPreparedStatementForValue(Number.class,ps).intValue();
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            this.getManager().close(ps);
            this.freeConnection(c);
            builder = null;
            sqlWhere = null;
        }
    }
    /**
     * fills the given StringBuilder with the sql where clauses constructed using the bean and the search type
     * @param sqlWhere the StringBuilder that will be filled
     * @param bean the bean to use for creating the where clauses
     * @param searchType exact ?  like ? starting like ?
     * @return the number of clauses returned
     */
    private int fillWhere(StringBuilder sqlWhere, B bean, int searchType)
    {
    	if (bean == null) {
    		return 0;
    	}
    	int dirtyCount = 0;
    	String sqlEqualsOperation = searchType == SEARCH_EXACT ? "=" : " like ";
    	List<String> fields = metaData.columnNames;
    	for(int i=0; i<fields.size(); ++i){
    		if(bean.isModified(i)){
    			++dirtyCount;
    			String col = metaData.columnNameOf(i);
    			if(bean.getValue(i) == null){
    				sqlWhere.append((sqlWhere.length() == 0) ? " " : " AND ").append(col).append(" IS NULL");
    			}else{
    				if(metaData.columnTypeOf(i) == String.class){
    					sqlWhere.append((sqlWhere.length() == 0) ? " " : " AND ").append(col).append(" ").append(sqlEqualsOperation).append("?");
    				} else{
    					sqlWhere.append((sqlWhere.length() == 0) ? " " : " AND ").append(col).append(" = ?");
    				}
    			}
    		}
    	}

    	return dirtyCount;
    }
    /**
     * fill the given prepared statement with the bean values and a search type
     * @param ps the PreparedStatement that will be filled
     * @param bean the bean to use for creating the where clauses
     * @param searchType exact ?  like ? starting like ?
     * @param fillNull wether fill null for null field
     * @param checkModified check column modified flag if true
     * @param fieldList  table of the field's associated constants,if null as all field
     * @param dirtyCount 
     * @return the number of clauses returned
     */
    private int fillPreparedStatement(PreparedStatement ps, BaseBean bean, int searchType,boolean fillNull, boolean checkModified, int[] fieldList, int dirtyCount) throws DaoException
    {
        return getManager().fillPreparedStatement(metaData, ps, bean, searchType, fillNull, checkModified, fieldList, dirtyCount);
    }
    
    /**
     * Retrieves an list of R using the junction table junction table(junctionTablename), given a linked table bean, 
     * specifying the start row and the number of rows.
     * @param rightType
     * @param left
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @param <L>
     * @param <R>
     * @return a list of R
     */
	public <L extends BaseBean,R extends BaseBean>
    List<R> loadViaJunctionTableAsList( Class<R> rightType,L left, int startRow, int numRows)
    {
    	checkState(!metaData.getJunctionTablePkMap().isEmpty(),"%s is not junction table",metaData.tablename);
    	checkArgument(null != left,"left is null");
    	checkArgument(null != rightType,"rightType is null");
        RowMetaData leftMetaData = RowMetaData.getMetaData(left.tableName());
        RowMetaData rightMetaData = RowMetaData.getMetaData(rightType);
        checkArgument(!leftMetaData.equals(rightMetaData),"same metadata  FOR left AND right type");
        checkArgument(metaData.isLinkedTable(leftMetaData.tablename),"INVALID left type %s",left.getClass());
        checkArgument(metaData.isLinkedTable(rightMetaData.tablename),"INVALID right type %s",rightType);
		Map<String, String> leftFields = metaData.junctionMapOf(leftMetaData.tablename);
    	Map<String, String> rightFields = metaData.junctionMapOf(rightMetaData.tablename);
        if(Iterables.tryFind(leftFields.values(), c->null == left.getValue(c)).isPresent()){
            return Collections.emptyList();
        }
        Connection c = null;
        PreparedStatement ps = null;
        String sql = " SELECT " + rightMetaData.columnFullFields
                        + " FROM "+ metaData.tablename + ", " + rightMetaData.tablename
                        + " WHERE "
                        + Joiner.on(" AND ").join(Iterables.transform(leftFields.keySet(), f->f + "=?"))
                        + " AND "
                        + Joiner.on(" AND ").join(Iterables.transform(rightFields.entrySet(), input->null == input ? "" : input.getKey() + "=" + input.getValue()));
        
        try
        {
            c = this.getConnection();
            PageQueryImplType pageQueryImplType = getManager().getPageQueryImplType();
            String wrapped = pageQueryImplType.wrap(sql, startRow, numRows);
            sql = firstNonNull(wrapped, sql);
            ps = getManager().getStatementCache().prepareStatement(c, sql, false, debug, 
                    "loadViaJunctionTableAsList", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            int pos =1;
            for(String key:leftFields.keySet()){
            	String fullName = leftFields.get(key);
            	int columnId = leftMetaData.columnIDOf(fullName);
            	getManager().fillPreparedStatement(ps, pos++, 
            			left.getValue(columnId), 
            			leftMetaData.sqlTypeOf(columnId), leftMetaData.columnTypeNames.get(columnId));	
            }
            BaseTableManager<R> rightManager = baseManagerOf(rightMetaData.tablename);
            if(null != wrapped){
                return rightManager.loadByPreparedStatementAsList(ps, null, 1, -1);
            }else{
                return rightManager.loadByPreparedStatementAsList(ps, null, startRow, numRows);
            }
        }
        catch (SQLException e)
        {
            throw new RuntimeDaoException(new DaoException(e.getMessage(), e));
        }
        finally
        {
           this.getManager().close(ps);
           this.freeConnection(c);
        }
    }

    private <L extends BaseBean,R extends BaseBean> B makeJunctionBean(L left,R right){
        Map<String, Object[]> map = metaData.getJunctionTablePkMap();
        checkState(!map.isEmpty(), "%s is not junction table",metaData.tablename);
        Object[] primaryValues = new Object[metaData.primaryKeyCount];
		for(int i=0; i<primaryValues.length; ++i){
			String pk = metaData.columnNameOf(i);
			Object[] data = map.get(pk);
			checkArgument(data!=null,"PRIMARY KEY %s NOT DEFINED",pk);
			String linkedTableName = (String)data[0];
			int linkedColumnId = ((Integer)data[1]).intValue();

			if(left.tableName().equals(linkedTableName)){
				primaryValues[i] = left.getValueChecked(linkedColumnId);
			}else if (right.tableName().equals(linkedTableName)){
				primaryValues[i] = right.getValueChecked(linkedColumnId);
			} else {
				throw new IllegalArgumentException(String.format("%s NOT linked table  %s", linkedTableName,metaData.tablename));
			}
		}
		return createBean(primaryValues);
    }

    /**
     * add junction between L and R if junction not exists
     * @param <L>
     * @param <R>
     * @param left
     * @param right
     */    
    private <L extends BaseBean,R extends BaseBean> 
    void addJunction(L left,R right){
        if(hasNullPk(left) || hasNullPk(right)){
            return ;
        }
        B junction = makeJunctionBean(left,right);
        if(!existsByPrimaryKey(junction)){
            save(junction);
        }
    }

    /**
     * remove junction between L and R 
     * @param <L>
     * @param <R>
     * @param left
     * @param right
     */    
    private <L extends BaseBean,R extends BaseBean> 
    int deleteJunction(L left,R right){
        if(hasNullPk(left) || hasNullPk(right)){
            return 0;
        }
        B junction = makeJunctionBean(left,right);
        return delete(junction);
    }
    
    @SuppressWarnings("unchecked")
    private <L extends BaseBean,R extends BaseBean>
    void addJunction(L left,R... rights){
        if(null != rights){
            for(R linked:rights){
                addJunction(left,linked);
            }
        }
    }

    private <L extends BaseBean,R extends BaseBean> 
    void addJunction(L left,Collection<R> rights){
        if(null != rights){
            for(R linked:rights){
                addJunction(left,linked);
            }
        }
    }
    @SuppressWarnings("unchecked")
    private <L extends BaseBean,R extends BaseBean>
    int deleteJunction(L left,R... rights){
        if(null != rights){
        	return deleteJunction(left, Arrays.asList(rights));
        }
        return 0;
    }
    private <L extends BaseBean,R extends BaseBean>
    int deleteJunction(L left,Collection<R> rights){
        int count = 0;
        if(null != rights){
            for(R right:rights){
                count += deleteJunction(left,right);
            }
        }
        return count;
    }
    /**
     * add junction between B and R if junction not exists
     * @param <R>
     * @param junction junction table name
     * @param bean
     * @param linked
     */    
	protected <R extends BaseBean> 
    void addJunction(String junction,B bean,R linked){
    	baseManagerOf(junction).addJunction(bean,linked);
    }
    /**
     * add junction between B and R if junction not exists
     * @param <R>
     * @param junction junction table name
     * @param bean
     * @param linkedBeans
     */    
    @SuppressWarnings({ "unchecked" })
	protected <R extends BaseBean> 
    void addJunction(String junction,B bean,R... linkedBeans){
    	baseManagerOf(junction).addJunction(bean,linkedBeans);
    }
    /**
     * add junction between B and R if junction not exists
     * @param <R>
     * @param junction junction table name
     * @param bean
     * @param linkedBeans
     */    
    @SuppressWarnings({ })
	protected <R extends BaseBean> 
    void addJunction(String junction,B bean,Collection<R> linkedBeans){
    	baseManagerOf(junction).addJunction(bean,linkedBeans);
    }
    /**
     * delete junction between B and R 
     * @param <R>
     * @param junction junction table name
     * @param bean
     * @param linked
     * @return count of deleted rows
     */    
	protected <R extends BaseBean> 
    int deleteJunction(String junction,B bean,R linked){
    	return baseManagerOf(junction).deleteJunction(bean,linked);
    }
    /**
     * delete junction between B and R 
     * @param <R>
     * @param junction junction table name
     * @param bean
     * @param linkedBeans
     * @return count of deleted rows
     */    
    @SuppressWarnings({ "unchecked" })
	protected <R extends BaseBean> 
    int deleteJunction(String junction,B bean,R... linkedBeans){
    	return baseManagerOf(junction).deleteJunction(bean,linkedBeans);
    }
    /**
     * delete junction between B and R 
     * @param <R>
     * @param junction junction table name
     * @param bean
     * @param linkedBeans
     * @return count of deleted rows
     */    
	protected <R extends BaseBean> 
    int deleteJunction(String junction,B bean,Collection<R> linkedBeans){
    	return baseManagerOf(junction).deleteJunction(bean,linkedBeans);
    }
    //28-2
    /** decode a resultset and call action
     * @param rs the resultset to decode
     * @param fieldList table of the field's associated constants
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @param action interface obj for do something
     * @return the count dealt by action  
     * @throws IllegalArgumentException
     */
    protected int actionOnResultSet(ResultSet rs, int[] fieldList, int startRow, int numRows, Action<B> action) throws DaoException{
        int count = 0;
        try{
            if(0!=numRows){
            	checkArgument(startRow>=1,"invalid argument:startRow (must >=1)");
            	checkArgument(null !=action && null != rs,"invalid argument:action OR rs (must not be null)");
           		for(;startRow>1&&rs.next();--startRow);//skip to last of startRow,can't use 'absolute' method for TYPE_FORWARD_ONLY
                if (fieldList == null) {
                    if(numRows<0){
                        for(;rs.next();++count){
                            action.call(decodeRow(rs));
                        }
                    }else{
                        for(;rs.next() && count<numRows;++count){
                            action.call(decodeRow(rs));
                        }
                    }
                }else {
                    if(numRows<0){
                        for(;rs.next();++count){
                            action.call(decodeRow(rs, fieldList));
                        }
                    }else{
                        for(;rs.next() && count<numRows;++count){
                            action.call(decodeRow(rs, fieldList));
                        }
                    }
                }
            }
            return count;
        }catch (QueueTimeoutException e) {
            SimpleLog.log(e.getClass().getSimpleName() + ":" + e.getMessage());
            return count;
        }catch(DaoException e){
            throw e;
        }catch(SQLException e){
            throw new DataAccessException(e);
        }
    }

	protected Object getColumnValue(ResultSet resultSet, int columnId) throws SQLException {
		ColumnCodec codec = metaData.columnCodecOf(columnId);
		if(null != codec) {
			Object obj = getManager().getObject(resultSet,columnId + 1, metaData.jdbcTypeOf(columnId));
			/*
			 * 如果字段值与字段类型匹配，直接返回
			 * 比如:authkernel ProductBean.sand字段,成员(field)类型为String,getter/setter方法类型为ByteBuffer.
			 * getObject根据JDBC类型返回的对象就是ByteBufer.此时如果再执行ByteBuffer到String的转换就会抛出异常，
			 * 也多此一举
			 */
			if(metaData.columnTypeOf(columnId).isInstance(obj)) {
				return obj;
			}
			return codec.deserialize(obj, metaData.fieldOf(columnId).getGenericType());
		}
		/** 这里是否也应该用 metaData.columnTypeOf(columnId)作为目标类型？待观察 */
		return getManager().getObject(resultSet,columnId + 1, metaData.fieldTypeOf(columnId));
	}
	protected void setColumnValue(B bean,int columnId, Object value ){
		 ColumnCodec codec = metaData.columnCodecOf(columnId);
		 if(null != codec) {
			 value = codec.deserialize(value,metaData.columnTypeOf(columnId));
		 }
		bean.setValue(columnId, value);
	}
    //29
    /**    
     * Transforms a ResultSet iterating on a B bean.
     *
     * @param rs the ResultSet to be transformed
     * @return bean resulting B bean
     */
    private B decodeRow(ResultSet rs) throws DaoException
    {
        B   bean = createBean();
        try
        {
        	for(int i=0; i<metaData.columnNames.size(); ++i){
        		bean.setValue(i, getColumnValue(rs, i));
        	}
        }
        catch(SQLException e)
        {
            throw new DataAccessException(e);
        }
        bean.setNew(false);
        bean.resetIsModified();

        return bean;
    }

    //30
    /**
     * Transforms a ResultSet iterating on a B bean according to a list of fields.
     *
     * @param rs the ResultSet to be transformed
     * @param fieldList table of the field's associated constants
     * @return bean resulting B bean
     */
    private B decodeRow(ResultSet rs, int[] fieldList) throws DaoException
    {
    	fieldList = MoreObjects.firstNonNull(fieldList, metaData.defaultColumnIdList);
        B bean = createBean();
        try
        {
            for(int i = 0; i < fieldList.length; i++)
            {
            	if(fieldList[i]>=0 && fieldList[i]<metaData.columnNames.size()){
            		int columnId = fieldList[i];
            		bean.setValue(columnId, getColumnValue(rs, columnId));
//            		setColumnValue(bean, columnId, getColumnValue(rs, columnId) );
            	}else{
            		throw new DaoException("Unknown field id " + fieldList[i]);
            	}
            }
        }
        catch(SQLException e)
        {
            throw new DataAccessException(e);
        }
        bean.setNew(false);
        bean.resetIsModified();

        return bean;
    }
    //////////////////////////////////////
    // PREPARED STATEMENT LOADER
    //////////////////////////////////////

    //34-1
    /**
     * Loads all the elements using a prepared statement specifying a list of fields to be retrieved,
     * and specifying the start row and the number of rows.
     *
     * @param ps the PreparedStatement to be used
     * @param fieldList table of the field's associated constants
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @return an array of B
     */
    protected List<B> loadByPreparedStatementAsList(PreparedStatement ps, int[] fieldList, int startRow, int numRows) throws DaoException
    {
    	ListAction<B> action = new ListAction<>();
        loadByPreparedStatement(ps,fieldList,startRow,numRows,action);
        return action.getList();
    }
    //34-2
    /**
     * Loads each element using a prepared statement specifying a list of fields to be retrieved,
     * and specifying the start row and the number of rows 
     * and dealt by action.
     *
     * @param ps the PreparedStatement to be used
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @param fieldList table of the field's associated constants
     * @param action Action object for do something(not null)
     * @return the count dealt by action
     */     
    private int loadByPreparedStatement(PreparedStatement ps, int[] fieldList, int startRow, int numRows,Action<B> action) throws DaoException
    {
        ResultSet rs =  null;
        try { 
            ps.setFetchSize(1000);
            rs = ps.executeQuery();
            return this.actionOnResultSet(rs, fieldList, startRow, numRows, action);
        } catch (DaoException e) {
            throw e;
        } catch (SQLException e) {
            throw new DataAccessException(e);
        } finally {
            this.getManager().close(rs);
        }
    }
    //_____________________________________________________________________
    //
    // LISTENER
    //_____________________________________________________________________    

    /** foreign key listener for DEELTE RULE */
    protected class  DeleteRuleListener<F extends BaseBean> extends BaseForeignKeyListener<F,B>{
    	protected final String fkName;
    	private final ForeignKeyRule deleteRule;
    	private final int[] foreignKeyIds;
    	DeleteRuleListener(String fkName) {
    		checkArgument(metaData.foreignKeys.containsKey(fkName),"INVALID foreign key name %s",fkName);
    		this.fkName = fkName;
    		this.deleteRule = metaData.foreignKeys.get(fkName).deleteRule;
    		this.foreignKeyIds = metaData.foreignKeyIdArrayOf(fkName);
    	}
    	@Override
    	protected List<B> getImportedBeans(F bean) {
    		return getListenerContainer().isEmpty() 
    				? Collections.<B>emptyList()
    				: loadByForeignKeyAsList(fkName,bean,1,-1);
    	}
		@SuppressWarnings("unchecked")
		@Override
		protected void onRemove(List<B> affectedBeans) throws DaoException {
			switch (deleteRule) {
			case SET_NULL:
				for(B bean : affectedBeans){
					BaseRow updated = ((BaseRow)bean).clone();
					for(int i=0; i<foreignKeyIds.length; ++i){
						updated.setValue(foreignKeyIds[i], null);
					}
					fire(Event.UPDATE, (B) updated);
				}
				break;
			case SET_DEFAULT:
				B tmpl = createBean();
				for(B bean : affectedBeans){
					BaseRow updated = ((BaseRow)bean).clone();
					for(int i=0; i<foreignKeyIds.length; ++i){
						updated.setValue(foreignKeyIds[i], tmpl.getValue(foreignKeyIds[i]));
					}
					fire(Event.UPDATE, (B) updated);
				}
				break;
			default:
				if(!deleteRule.isNoAction()){
					Event event = Event.valueOf(deleteRule.eventOfDeleteRule);
					for(B bean : affectedBeans){
						fire(event, bean);
					}
				}
				break;
			}
		}
		
		/**
		 * fire current event by  {@link TableListener}
		 * @param event
		 * @param bean
		 * @throws RuntimeDaoException
		 */
		private void fire(Event event,B bean)throws RuntimeDaoException {

			ListenerContainer<B> container = getListenerContainer();
			switch(event){
			case UPDATE:
				container.beforeUpdate(bean);
				container.afterUpdate(bean);
				container.done();
				break;
			case DELETE:
				container.beforeDelete(bean);
				container.afterDelete(bean);
				container.done();
				break;
			case INSERT:
				// DO NOTHING
				// container.beforeInsert(bean);
				// container.afterInsert(bean);
				// container.done();
			default:
				break;
			}
		}

		@Override
		public String toString() {
			StringBuilder builder = new StringBuilder();
			builder.append("DeleteRuleListener [fkName=");
			builder.append(fkName);
			builder.append(", deleteRule=");
			builder.append(deleteRule);
			builder.append(", foreignKeyIds=");
			builder.append(metaData.columnNamesOf(foreignKeyIds));
			builder.append("]");
			return builder.toString();
		}	
    }

    //35

    @Override
    public void registerListener(TableListener<B> listener)
    {
        this.getListenerContainer().add(listener);
		if(debug){
			log("REGISTER TABLE LISTENER {}",listener);
		}
    }

    //36

    @Override
    public void unregisterListener(TableListener<B> listener)
    {
        this.getListenerContainer().remove(listener);
		if(debug){
			log("UNREGISTER TABLE LISTENER {}",listener);
		}
    }
    
    //37-2
    /**
     * bind foreign key listener to foreign table: <br>
     * DELETE RULE <br>
     */
    public void bindForeignKeyListenerForDeleteRule(){
    	for(Entry<String, TableListener<BaseBean>> entry : getForeignKeyDeleteListeners().entrySet()){
    		TableManager<BaseBean> manager = managerOf(metaData.foreignKeys.get(entry.getKey()).foreignTable);
    		manager.registerListener(entry.getValue());
    	}
    }
    
  //37-3
    /**
     * unbind foreign key listener from all of foreign tables <br>
     * @see #bindForeignKeyListenerForDeleteRule()
     */
    public void unbindForeignKeyListenerForDeleteRule(){
    	for(Entry<String, TableListener<BaseBean>> entry : getForeignKeyDeleteListeners().entrySet()){
    		TableManager<BaseBean> manager = managerOf(metaData.foreignKeys.get(entry.getKey()).foreignTable);
    		manager.unregisterListener(entry.getValue());
    	}
    }
    //_____________________________________________________________________
    //
    // UTILS
    //_____________________________________________________________________
    //40
    /**
     * Retrieves the manager object used to get connections.
     *
     * @return the manager used
     */
    protected Manager getManager()
    {
        if(null == manager){
            synchronized (this) {
                if(null == manager){
                    Manager m = Managers.managerInstanceOfAlias(metaData.alias);
                    manager = m == null ? Manager.getInstance() : m;
                }
            }
        }
        return manager;
    }

    //41
    /**
     * Frees the connection.
     *
     * @param c the connection to release
     */
    protected void freeConnection(Connection c)
    {
        // back to pool
        this.getManager().releaseConnection(c);
    }

    //42
    /**
     * Gets the connection.
     */
    protected Connection getConnection() throws DaoException
    {
        try
        {
            return this.getManager().getConnection();
        }
        catch(SQLException e)
        {
            throw new DataAccessException(e);
        }
    }
    
    private int loadBySqlForAction(String sql, Object[] argList, int[] fieldList,int startRow, int numRows,Action<B> action){
        PreparedStatement ps = null;
        Connection c = null;
        try {
            c = this.getConnection();
            AtomicLong count = new AtomicLong(-1L);
            /** 先对SQL执行注入攻击检查及归一化处理 */
            sql = getManager().getStatementCache().normalize(sql, true);
            String wrapped = getManager().rebuildSelectSql(c,sql,argList, startRow, numRows, count, debug);
            // 执行count语句返回0,就不必再继续执行SQL查询
            if(0 == count.get()){
                log(debug,"count 0,SKIP execute sql");
            	return 0;
            }
            sql = firstNonNull(wrapped, sql);
            ps = getManager().getStatementCache().prepareStatement(c, sql, false, debug, 
                    "loadBySqlForAction", ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            getManager().fillPrepareStatement(ps, argList);
            if(null != wrapped){
                return this.loadByPreparedStatement(ps, fieldList, 1, -1, action);
            }else{
                return this.loadByPreparedStatement(ps, fieldList, startRow, numRows, action);
            }
        } catch (SQLException e) {
            throw new RuntimeDaoException(new DataAccessException(e));
        } finally {
            this.getManager().close(ps);
            this.freeConnection(c);
        }
    }
    @Override
	public List<BaseBean> runSqlAsList(Map<String, Class<?>> targetTypes, String sql, Object... argList) {
    	return getManager().runSqlAsList(targetTypes,sql, argList);
	}

	@Override
	public List<BaseBean> runSqlAsList(String sql, Object... argList) throws RuntimeDaoException{
    	return getManager().runSqlAsList(sql, argList);
    }
    
    @Override
	public List<Map<String, Object>> runSqlForMap(Map<String,Class<?>> targetTypes, String sql,Object... argList) throws RuntimeDaoException{
    	return getManager().runSqlForMap(targetTypes, sql, argList);
    }
    
    @Override
	public <T>List<T> runSqlAsList(Class<T> targetType, String sql,Object... argList) throws RuntimeDaoException{
    	return getManager().runSqlAsList(targetType, sql, argList);
    }
    
    @Override
	public <T> T runSqlForValue(Class<T> targetType,String sql, Object... argList) throws RuntimeDaoException{
    	return getManager().runSqlForValue(targetType, sql, argList);
    }
    @Override
	public boolean runSql(String sql, Object[] argList) {
    	return getManager().runSql(sql, argList);
    }
    @Override
    public int runSql(String sql) {
    	return getManager().runSql(sql);
    }
    @Override
    public <T>T runAsTransaction(Callable<T> fun) {
       return getManager().runAsTransaction(fun);
    }
    
    @Override
    public void runAsTransaction(Runnable fun) {
       	getManager().runAsTransaction(fun);
    }    
    
	@Override
	public <T> T runWithNoPage(Callable<T> fun) throws RuntimeDaoException {
		return getManager().runWithNoPage(fun);
	}

	@Override
	public void runWithNoPage(Runnable fun) throws RuntimeDaoException {
		getManager().runWithNoPage(fun);
	}
    //45 

    @Override
    public long rowCountOf(String sql) throws RuntimeDaoException {
        return getManager().rowCountOf(sql);
    }

    @Override
	public GeometryDataCodec getGeometryDataCodec() {
    	return getManager().getGeometryDataCodec();
    }
    /**
     * @param <T> PK type
     * @param type
     * @param beans
     * @return return a primary key list from B array
     * @see #toPrimaryKeyList(Class,Collection)
     */
    @SuppressWarnings("unchecked")
	protected <T> List<T> toPrimaryKeyList(Class<T>type,B... beans){        
        if(null == beans || beans.length ==0){
            return Collections.emptyList();
        }
        return toPrimaryKeyList(type,Arrays.asList(beans));
    }
    
    /**
	 * listener event:<br>
	 * {@code INSERT} insert a bean<br>
	 * {@code UPDATE} update a bean<br>
	 * {@code DELETE} delete a bean<br>
	 * @author guyadong
	 *
	 */
	public static enum Event{        
	    /** insert a bean */INSERT,
	    /** update a bean */UPDATE,
	    /** delete a bean */DELETE
	}

	//46 
    /**
     * return a primary key list from B collection<br>
     * throw {@link UnsupportedOperationException} if there is more than a primary key
     * @param <T> PK type
     * @param type PK type
     * @param beans input beans
     */
    protected <T> List<T> toPrimaryKeyList(Class<T>type,Collection<B> beans){
        if(metaData.primaryKeyNames.length != 1){
        	throw new UnsupportedOperationException();
        }
        if(null == beans){
            return Collections.emptyList();
        }
        int pkId = metaData.primaryKeyIds[0];
        checkArgument(metaData.columnTypeOf(pkId).equals(type),"INVALID primary key type: " + type);
        return Lists.newArrayList(Iterables.transform(beans, input->input == null ? null : input.getValue(pkId)));
    }
    private Object[] toPkValues(B bean,int[] selfFkIds){
    	checkArgument(selfFkIds.length == metaData.primaryKeyNames.length,"MISMATCH SIZE of primary keys");
    	return bean.asValueArray(selfFkIds);
    }
    /**
     * return bean list ( include bean specified by {@code primaryKeys} ) by the self-reference field specified by fkName<br>
     * first element is top bean
     * @param fkName foreign key name
     * @param primaryKeys values of primary keys 
     * @return  empty list if input primary key is {@code null}<br>
     *         first element equal last if self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected List<B> listOfSelfRef(String fkName,Object... primaryKeys) throws RuntimeDaoException{
        List<B> list = new ArrayList<>();
        int[] selfFks = metaData.foreignKeyIdArrayOf(fkName); 
        for(B ref = loadByPrimaryKey(primaryKeys)
                ; null != ref
                ; ref = loadByPrimaryKey(toPkValues(ref,selfFks))){
            list.add(ref);
            Object[] refPk = ref.primaryValues();
            Object[] refSelf = toPkValues(ref,selfFks);
            
            if(Arrays.equals(refPk,refSelf)
            		|| (list.size() > 1 && Arrays.equals(refPk,primaryKeys))){
                // cycle reference
                break;
            }
        }
        Collections.reverse(list);
        return list;
    }
    /**
     * return bean list ( include {@code bean} ) by the self-reference field specified by fkName<br>
     * first element is top bean
     * @param fkName foreign key name
     * @param bean 
     * @return  empty list if input primary key is {@code null}<br>
     *         first element equal last if self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected List<B> listOfSelfRef(String fkName,B bean) throws RuntimeDaoException{
    	return bean == null ? Collections.<B>emptyList() : listOfSelfRef(fkName,bean.primaryValues());
    }
    /**
     * get level count on the self-reference field specified by fkName
     * @param fkName foreign key name
     * @param primaryKeys values of primary keys 
     * @return  0 if input primary key is {@code null}<br>
     *         -1 if self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected int levelOfSelfRef(String fkName,Object... primaryKeys) throws RuntimeDaoException{
        int count = 0 ;
        int[] selfFks = metaData.foreignKeyIdArrayOf(fkName); 
        for(B ref = loadByPrimaryKey(primaryKeys)
           ; null != ref
           ; ++count,ref = loadByPrimaryKey(toPkValues(ref,selfFks))){
            Object[] refPk = ref.primaryValues();
            Object[] refSelf = toPkValues(ref,selfFks);
            if(    (Arrays.equals(refPk,refSelf))
                || (count > 0 && Arrays.equals(refPk,primaryKeys))){
                // cycle reference
                return -1;
            }
        }
        return count;
    }
    /**
     * get level count on the self-reference field specified by fkName
     * @param fkName foreign key name
     * @param bean 
     * @return  0 if input primary key is {@code null}<br>
     *         -1 if self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected int levelOfSelfRef(String fkName,B bean) throws RuntimeDaoException{
    	return bean == null ? 0 : levelOfSelfRef(fkName,bean.primaryValues());
    }
    /**
     * test whether the self-reference field specified by fkName is cycle
     * @param fkName foreign key name
     * @param primaryKeys values of primary keys
     * @return true if the self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected boolean isCycleOfSelfRef(String fkName,Object... primaryKeys) throws RuntimeDaoException{
    	return levelOfSelfRef(fkName,primaryKeys) < 0;
    }
    /**
     * test whether the self-reference field specified by fkName is cycle
     * @param fkName foreign key name
     * @param bean
     * @return true if the self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected boolean isCycleOfSelfRef(String fkName,B bean) throws RuntimeDaoException{
    	return bean == null ? false : levelOfSelfRef(fkName,bean.primaryValues()) < 0;
    }
    /**
     * Ensures the self-reference field specified by fkName is not cycle
     * @param fkName foreign key name
     * @param primaryKey 
     * @return always {@code primaryKey}
     * @throws IllegalStateException if self-reference field is cycle 
     * @throws RuntimeDaoException
     * @see #isCycleOfSelfRef(String,Object[])
     */
    protected <T>T checkCycleOfSelfRef(String fkName,T primaryKey)
    		throws IllegalStateException,RuntimeDaoException{
        if(isCycleOfSelfRef(fkName,primaryKey)){
            throw new IllegalStateException("cycle on foreign key: " + fkName);
        }
        return primaryKey;
    }
    /**
     * Ensures the self-reference field specified by fkName is not cycle
     * @param fkName foreign key name
     * @param bean
     * @return always {@code bean}
     * @throws IllegalStateException if self-reference field is cycle 
     * @throws RuntimeDaoException
     * @see #isCycleOfSelfRef(String, Object...)
     */
    protected B checkCycleOfSelfRef(String fkName,B bean)
    	throws IllegalStateException,RuntimeDaoException{
        if(isCycleOfSelfRef(fkName,bean)){
            throw new IllegalStateException("cycle on foreign key: " + fkName);
        }
        return bean;
    }
    /**
     * return B bean that with {@code null} self-reference field specified by fkName
     * @param fkName foreign key name
     * @param primaryKeys values of primary keys
     * @return top B instance,or self bean if not exist top bean
     * @throws IllegalArgumentException if input primary keys has {@code null}
     * @throws IllegalStateException if self-reference field is cycle
     * @throws ObjectRetrievalException not found record by primary key
     * @throws RuntimeDaoException
     */
    protected B topOfSelfRef(String fkName,Object... primaryKeys) 
    		throws IllegalArgumentException,IllegalStateException,ObjectRetrievalException,RuntimeDaoException{
        checkArgument(!hasNull(primaryKeys),"primaryKeys has null element");
        int[] selfFks = metaData.foreignKeyIdArrayOf(fkName); 

        B ref = loadByPrimaryKeyChecked(primaryKeys);
        int count = 0 ;
        Object[] refSelf;
        for(;!hasNull(refSelf = toPkValues(ref,selfFks));){
            Object[] refPk = ref.primaryValues();
        	 // cycle reference
        	checkState(!(Arrays.equals(refPk,refSelf)
                || (++count > 1 && Arrays.equals(refPk,primaryKeys))),"cycle on fk: " + fkName);
            ref = loadByPrimaryKeyChecked(refSelf);
        }
        return ref;
    }
    /**
     * return B bean that with {@code null} self-reference field specified by fkName 
     * @param fkName foreign key name
     * @param bean  
     * @return B instance
     * @throws IllegalArgumentException if input primary keys has {@code null}
     * @throws IllegalStateException if self-reference field is cycle
     * @throws ObjectRetrievalException not found record by primary key
     * @throws RuntimeDaoException
     */
    protected B topOfSelfRef(String fkName,B bean)
    		throws IllegalArgumentException,IllegalStateException,ObjectRetrievalException,RuntimeDaoException{
        return topOfSelfRef(fkName,checkNotNull(bean,"bean is null").primaryValues());
    }
    private LinkedHashSet<B> doListOfChild(String fkName, B bean,LinkedHashSet<B> set){
        bean = loadByPrimaryKey(bean);
        if(null != bean){
            checkState(!set.contains(bean),"cycle on foreign key: " + fkName);
            set.add(bean);
            List<B> childs = loadByForeignKeyAsList(fkName, bean, 1, -1);
            for(B c:childs){
                doListOfChild(fkName,c,set);
            }
        }
        return set;
    }
    /**
     * return child bean list (self included) by the self-reference field specified by fkName <br>
     * throw {@link RuntimeDaoException} if self-reference field is cycle
     * @param fkName foreign key name
     * @param bean
     * @return  child bean list,empty list if not found record
     * @throws IllegalStateException if self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected List<B> childListOfSelfRef(String fkName, B bean)
    		 throws IllegalStateException,RuntimeDaoException{
    	return new ArrayList<B>(doListOfChild(fkName,bean,new LinkedHashSet<B>()));
    }
    /**
     * return child bean list (self included) by the self-reference field specified by fkName <br>
     * throw {@link RuntimeDaoException} if self-reference field is cycle
     * @param fkName foreign key name
     * @param primaryKeys values of primary keys
     * @return  child bean list,empty list if not found record
     * @throws IllegalStateException if self-reference field is cycle
     * @throws RuntimeDaoException
     */
    protected List<B> childListOfSelfRef(String fkName, Object... primaryKeys) 
    		throws IllegalStateException,RuntimeDaoException{
    	return childListOfSelfRef(fkName,createBean(primaryKeys));
    }

    /**
	 * return child bean list (self included) by the self-reference field specified by fkName <br>
	 * throw {@link RuntimeDaoException} if self-reference field is cycle
	 * @param fkName foreign key name
	 * @param bean
	 * @return  child bean list,empty list if not found record
	 * @throws IllegalStateException if self-reference field is cycle
	 * @throws RuntimeDaoException
	 * @since 3.23.0
	 */
	protected LinkedHashSet<B> childrenOfSelfRef(String fkName, B bean)
			throws IllegalStateException,RuntimeDaoException{
		return doListOfChild(fkName,bean,new LinkedHashSet<B>());
	}

    /**
     * return child bean list (self included) by the self-reference field specified by fkName <br>
     * throw {@link RuntimeDaoException} if self-reference field is cycle
     * @param fkName foreign key name
     * @param primaryKeys values of primary keys
     * @return  child bean list,empty list if not found record
     * @throws IllegalStateException if self-reference field is cycle
     * @throws RuntimeDaoException
     * @since 3.23.0
     */
    protected LinkedHashSet<B> childrenOfSelfRef(String fkName, Object... primaryKeys) 
    		throws IllegalStateException,RuntimeDaoException{
    	return childrenOfSelfRef(fkName,createBean(primaryKeys));
    }
    /**
     * return child bean list (self included) by the self-reference field specified by fkName <br>
     * throw {@link RuntimeDaoException} if self-reference field is cycle
     * @param fkName foreign key name
     * @param beans B beans with primary keys
     * @return  child bean list,empty list if not found record
     * @throws IllegalStateException if self-reference field is cycle
     * @throws RuntimeDaoException
     * @since 3.23.0
     */
    protected LinkedHashSet<B> childrenOfBeansOfSelfRef(String fkName, Iterable<B> beans)
    		throws RuntimeDaoException{
    	if(1 != metaData.primaryKeyCount) {
    		throw new UnsupportedOperationException(String.format("UNSUPPORT %s,one primary key requied",fkName));
    	}
    	LinkedHashMap<Object,B> children = Maps.newLinkedHashMap();
    	if(null != beans && !Iterables.isEmpty(beans)) {
    		for(B bean:Sets.newLinkedHashSet(Iterables.filter(beans, Predicates.notNull()))) {
    			Object pk = bean.getValue(metaData.primaryKeyIds[0]);
    			if(null != pk && !children.containsKey(pk)) {
    				children.putAll(
    						Maps.uniqueIndex(childrenOfSelfRef(fkName,bean),
    								new Function<B,Object>(){
		    							@Override
		    							public Object apply(B input) {
		    								return input.getValue(metaData.primaryKeyIds[0]);
		    							}}));
    			}
    		}
    	}
    	return Sets.newLinkedHashSet(children.values());
    }
    /**
     * return child bean list (self included) by the self-reference field specified by fkName <br>
     * throw {@link RuntimeDaoException} if self-reference field is cycle
     * @param fkName foreign key name
     * @param primaryKeys values of primary keys
     * @return  child bean list,empty list if not found record
     * @throws IllegalStateException if self-reference field is cycle
     * @throws RuntimeDaoException
     * @since 3.23.0
     */
    protected <T> LinkedHashSet<B> childrenOfPksOfSelfRef(String fkName, Iterable<T> primaryKeys)
    		throws RuntimeDaoException{
    	if(1 != metaData.primaryKeyCount) {
    		throw new UnsupportedOperationException(String.format("UNSUPPORT %s,one primary key requied",fkName));
    	}
    	LinkedHashMap<T,B> children = Maps.newLinkedHashMap();
    	if(null != primaryKeys && !Iterables.isEmpty(primaryKeys)) {
    		for(T pk:Sets.newLinkedHashSet(Iterables.filter(primaryKeys, Predicates.notNull()))) {
    			if(!children.containsKey(pk)) {
    				children.putAll(
    						Maps.uniqueIndex(childrenOfSelfRef(fkName,pk),
    								new Function<B,T>(){
		    							@Override
		    							public T apply(B input) {
		    								return input.getValue(metaData.primaryKeyIds[0]);
		    							}}));
    			}
    		}
    	}
    	return Sets.newLinkedHashSet(children.values());
    }
	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((metaData == null) ? 0 : metaData.hashCode());
		return result;
	}

	@Override
	public boolean equals(Object obj) {
		if (this == obj) {
			return true;
		}
		if (obj == null) {
			return false;
		}
		if (getClass() != obj.getClass()) {
			return false;
		}
		BaseTableManager<?> other = (BaseTableManager<?>) obj;
		if (metaData == null) {
			if (other.metaData != null) {
				return false;
			}
		} else if (!metaData.equals(other.metaData)) {
			return false;
		}
		return true;
	}

	@Override
	public String toString() {
		StringBuilder builder = new StringBuilder();
		builder.append(getClass().getSimpleName() + " [metaData=");
		builder.append(metaData);
		builder.append("]");
		return builder.toString();
	}

	/**
	 * set debug flag that determine if output log message,default : false
	 * @param debug flag for debug message output
	 */
	static void setDebug(boolean debug) {
		BaseTableManager.debug = debug;
	}

	/**
	 * @param whereCheckFlag
	 */
	public static void setWhereCheckFlag(int whereCheckFlag) {
	}

	/**
	 * @return debug flag 
	 */
	static boolean isDebug() {
		return debug;
	}
	
    public class DeleteBeanAction implements Action<B>{
        private final AtomicInteger count=new AtomicInteger(0);
        public DeleteBeanAction() {
		}
		@Override
        public void call(B bean){
                delete(bean);
                count.incrementAndGet();
        }
        public int getCount(){
            return count.get();
        }
    }

}